package com.yujiahao.bigdata.streaming

import org.apache.spark.SparkConf

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
//窗口数据的范围必须为采集周期的整数倍，步长也必须为采集周期的整数
//窗口在滑动的过程中，如果头结尾，尾接头，那么没有重复数据，称之为滚动窗口，窗口在滑动的过程中，窗口的范围和滑动幅度一样
//窗口在滑动的过程中，窗口数据有重复，因为窗口的范围大于窗口滑动幅度，这样的窗口称之为滑动窗口，窗口在滑动过程中窗口的范围小于滑动幅度就会丢数据
//如果窗口在滑动的过程中，窗口范围大，但是滑动幅度小，可能会有大量重复数据，数据在计算时候，性能就比较差，为了提供效率，可以采用特殊的的方法提升效率

object Stream_reduceByKeyAndWindow {
  def main(args: Array[String]): Unit = {
    //TODO 1、SparkStreaming环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //TODO 2、StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    //TODO 5、注意设置检查点，因为你每次只处理新添加的和要去除的，那么中间的数据你要保存到检查点中去
    ssc.checkpoint("cp")
    //TODO 5、监控某个服务器的数据
    val socektDS: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)
    //TODO 5.1、将采集的数据进行结构转换，以便后面的聚合计算
    val word: DStream[(String, Int)] = socektDS.map((_, 1))
    //TODO 5.2、这里使用窗口进行操作，注意要提前声明检查点，将中简的数据保存起来， 否则会报错
    word.reduceByKeyAndWindow(
      (x, y) => {//将新增的数据累加
      x + y
    },
      (x, y) => {
        x - y //将滑出去的数据减掉
      },Seconds(6),Seconds(3)).print()

    //TODO 3、启动采集器
    ssc.start()
    //TODO 4、Driver等待采集器的结束，否则，当前Driver处于阻塞状态
    ssc.awaitTermination()

  }

}
